package com.idanchuang.count.support;


import com.google.common.eventbus.Subscribe;
import com.idanchuang.count.core.entity.KafkaSendEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.kafka.core.KafkaTemplate;

/**
 * 这样设计是因为直接调用kafka.send()，虽然是异步发送，但是在高并发情况下，线程同步会有一定的性能损耗
 * 所以改成异步单线程发送，保证无锁，提高性能
 * @author Richard_yyf
 * @version 1.0 2021/9/26
 */
@Slf4j
public class KafkaAsyncSender implements DisposableBean {

    private final KafkaTemplate<Object, Object> kafkaTemplate;

    public KafkaAsyncSender(KafkaTemplate<Object, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @SuppressWarnings("UnstableApiUsage")
    @Subscribe
    public void send(KafkaSendEvent event) {

        try {
            if (event.getCallback() == null) {
                kafkaTemplate.send(event.getRecord());
            } else {
                kafkaTemplate.send(event.getRecord()).addCallback(event.getCallback());
            }
        } catch (Exception e) {
            // 打印错误日志触发告警，不做其他处理
            log.error("kafka异步发送异常", e);
        }
    }

    @Override
    public void destroy() throws Exception {
        //  程序关闭时会强制flush
        log.info("application exit, try flush kafka producer");
        kafkaTemplate.flush();
        log.info("application exit, kafka producer flush success");
    }
}
